﻿using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace RatingAuthor
{
  public static class BlockingCollectionExtensions
  {
    public static BlockingCollection<U> Process<T, U>(this BlockingCollection<T> src, Func<T, U> processor)
    {
      return new Processor<T, U>(src, processor).Result;
    }
    public static BlockingCollection<T> Process<T>(this BlockingCollection<T> src, Action<T> processor)
    {
      return new Processor<T, T>(src,
               i =>
               {
                 processor(i);
                 return i;
               }).Result;
    }
    public static BlockingCollection<T> SingleThreadedProcess<T>(this BlockingCollection<T> src, Action<T> processor)
    {
      return new SingleThreadedProcessor<T, T>(src,
               i =>
               {
                 processor(i);
                 return i;
               }).Result;
    }
    public static Task AsCollectingTask<T>(this BlockingCollection<T> src)
    {
      return Task.Factory.StartNew(
        () =>
        {
          while (!src.IsCompleted)
          {
            try
            {
              // Blocks if _src.Count == 0
              src.Take();
            }
            catch (InvalidOperationException)
            {
              // IOE means that Take() was called on a completed collection.
              // Some other thread can call CompleteAdding after we pass the
              // IsCompleted check but before we call Take. 
              // In this example, we can simply catch the exception since the 
              // loop will break on the next iteration.
            }
          }
          src.Dispose();
        });
    }

    private class Processor<T, U>
    {
      private readonly BlockingCollection<T> _src;
      private readonly Func<T, U> _processor;
      private readonly BlockingCollection<U> _result;
      private int _workersCount;

      public Processor(BlockingCollection<T> src, Func<T, U> processor)
      {
        _src = src;
        _processor = processor;
        _result = new BlockingCollection<U>();
        StartWorker();
      }


      private void StartWorker()
      {
        Interlocked.Increment(ref _workersCount);
        Task.Factory.StartNew(
          () =>
          {
            Worker();
            if (Interlocked.Decrement(ref _workersCount) == 0)
            {
              _result.CompleteAdding();
              _src.Dispose();
            }
          });

      }
      private void Worker()
      {
        if (_src.IsCompleted) return;
        try
        {
          // Blocks if _src.Count == 0
          var item = _src.Take();
          StartWorker();
          _result.Add(_processor(item));
        }
        catch (InvalidOperationException)
        {
          // IOE means that Take() was called on a completed collection.
          // Some other thread can call CompleteAdding after we pass the
          // IsCompleted check but before we call Take. 
          // In this example, we can simply catch the exception since the 
          // loop will break on the next iteration.
        }
      }

      public BlockingCollection<U> Result
      {
        get { return _result; }
      }
    }
    private class SingleThreadedProcessor<T, U>
    {
      private readonly BlockingCollection<T> _src;
      private readonly Func<T, U> _processor;
      private readonly BlockingCollection<U> _result;

      public SingleThreadedProcessor(BlockingCollection<T> src, Func<T, U> processor)
      {
        _src = src;
        _processor = processor;
        _result = new BlockingCollection<U>();
        Worker();
      }
      private void Worker()
      {
        while (!_src.IsCompleted)
        {
          try
          {
            // Blocks if _src.Count == 0
            _result.Add(_processor(_src.Take()));
          }
          catch (InvalidOperationException)
          {
            // IOE means that Take() was called on a completed collection.
            // Some other thread can call CompleteAdding after we pass the
            // IsCompleted check but before we call Take. 
            // In this example, we can simply catch the exception since the 
            // loop will break on the next iteration.
          }
        }
      }

      public BlockingCollection<U> Result
      {
        get { return _result; }
      }
    }
  }
}